package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.ExamCorrectPercentBean;
import com.atguigu.edu.realtime.common.bean.ExamPageBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.util.Iterator;
import java.util.List;

public class DwsExamCorrectPercentApp extends BaseApp {

    public static void main(String[] args) {
        new  DwsExamCorrectPercentApp().start(13003,4,"DwsExamCorrectPercentApp", Constant.TOPIC_DWD_EXAMINATION_QUESTION_DETAIL);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStrDS.map(JSON::parseObject);

        SingleOutputStreamOperator<JSONObject> wxStream = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(
                        (v, ts) -> v.getLong("ts") * 1000
                ));


        SingleOutputStreamOperator<ExamCorrectPercentBean> beanStream = wxStream.keyBy(v -> v.getString("id")).process(
                new ProcessFunction<JSONObject, ExamCorrectPercentBean>() {

                    ListState<String> listState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<String>("ListStateDescriptor", String.class);
                        listState = getRuntimeContext().getListState(listStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, ExamCorrectPercentBean>.Context context, Collector<ExamCorrectPercentBean> collector) throws Exception {
                        Long personNum = 0L;
                        Long correctPerson = 0L;
                        Long correctTimes = 0L;

                        //从jsonObject中获取 user_id ,在状态中查找，如果有，独立访客数为0，否则为1，再加入状态中
                        String userId = jsonObject.getString("user_id");

                        Iterator<String> iterator = listState.get().iterator();
                        List<String> userIds = IteratorUtils.toList(iterator);
                        if (!userIds.contains(userId)) {
                            //独立访客数为1
                            personNum = 1L;
                            listState.add(userId);
                            if ("1".equals(jsonObject.getString("is_correct"))) {
                                //正确答题的独立访客数为1
                                correctPerson = 1L;
                            }
                        }
                        if ("1".equals(jsonObject.getString("is_correct"))) {
                            //正确答题次数为1
                            correctTimes = 1L;
                        }
                        ExamCorrectPercentBean examCorrectPercentBean = ExamCorrectPercentBean
                                .builder()
                                .answerTimes(1L)
                                .correctTimes(correctTimes)
                                .correctPerson(correctPerson)
                                .personNum(personNum)
                                .paperId(jsonObject.getString("paper_id"))
                                .build();

                        collector.collect(examCorrectPercentBean);
                    }
                }
        );
        //beanStream.print();

        SingleOutputStreamOperator<ExamCorrectPercentBean> resultStream = beanStream.keyBy(v -> v.getPaperId()).window(
                TumblingProcessingTimeWindows.of(Time.seconds(10))
        ).reduce(
                new ReduceFunction<ExamCorrectPercentBean>() {
                    @Override
                    public ExamCorrectPercentBean reduce(ExamCorrectPercentBean examCorrectPercentBean, ExamCorrectPercentBean t1) throws Exception {
                        examCorrectPercentBean.setCorrectTimes(examCorrectPercentBean.getCorrectTimes() + t1.getCorrectTimes());
                        examCorrectPercentBean.setAnswerTimes(examCorrectPercentBean.getAnswerTimes() + t1.getAnswerTimes());
                        examCorrectPercentBean.setCorrectPerson(examCorrectPercentBean.getCorrectPerson() + t1.getCorrectPerson());
                        examCorrectPercentBean.setPersonNum(examCorrectPercentBean.getPersonNum() + t1.getPersonNum());
                        return examCorrectPercentBean;
                    }
                },
                new WindowFunction<ExamCorrectPercentBean, ExamCorrectPercentBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<ExamCorrectPercentBean> iterable, Collector<ExamCorrectPercentBean> collector) throws Exception {
                        ExamCorrectPercentBean next = iterable.iterator().next();
                        next.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        next.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        next.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(next);
                    }
                }
        );


        SingleOutputStreamOperator<ExamCorrectPercentBean> dimStream = resultStream.map(
                new RichMapFunction<ExamCorrectPercentBean, ExamCorrectPercentBean>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHbaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public ExamCorrectPercentBean map(ExamCorrectPercentBean orderBean) throws Exception {
                        //根据流中的对象获取要关联的维度的主键
                        String paperId = orderBean.getPaperId();
                        //根据维度的主键获取对应的维度对象
                        JSONObject dimJsonObj = HBaseUtil.readRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_test_paper", paperId, JSONObject.class);
                        //将维度对象相关的维度属性补充到流中的对象上
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setPaperName(dimJsonObj.getString("paper_title"));

                        return orderBean;
                    }
                }
        );

        dimStream.print();


        dimStream.map(new BeanToJsonStrMapFunction())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_exam_correct_percent_window"));
    }
}
